package com.udf.flink.demo.stream.state.checkpoint;

import com.udf.flink.demo.stream.function.MyStatefulMapFunction;
import com.udf.flink.demo.stream.source.SimpleCustomSource;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * FsStateBackend
 * Created by wy on 2020/12/9.
 */
public class RocksDBStateBackendExample {

    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(6000); // checkpoint 时间差
        env.setParallelism(1);
//        # 'env' is the createdStreamExecutionEnvironment
//# 'true' is to enable incrementalcheckpointing
//        env.setStateBackend(new RocksDBStateBackend("hdfs:///fink-checkpoints", true));
        EmbeddedRocksDBStateBackend rc = new EmbeddedRocksDBStateBackend();
        rc.setDbStoragePath("d:\\home");
        env.setStateBackend(rc);
//        env.getCheckpointConfig().setCheckpointStorage("file:///home/xuzh"); file system checkpoint
        System.out.println(rc.getDbStoragePaths());
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        DataStream<Tuple2<String, Integer>> source = env.addSource(new SimpleCustomSource()).uid("simple-custom-source");
        source.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.getField(0);
            }
        }).map(new MyStatefulMapFunction()).uid("my-stateful-map-function").print();

        env.execute("fsStateBackend-example");
    }
}
